"""
Case Type   : 子事务并发回滚功能模块-5.1.0新增特性
Case Name   : 不带子事务的多个事务并发执行过程进行kill -9
Create At   : 2024/1/24
Owner       : peilinqian
Description :
    1. 创建表
    2. 创建存储过程
    3. 初始查询ProcArrayLock锁的wait情况
    4. 并发执行存储过程
    5. kill-9 执行事务
    6. 再次查询ProcArrayLock锁的wait情况
    7. 验证并发事务执行结果
    8. 并发执行存储过程
    9. 验证并发事务执行结果
    10. 再次查询ProcArrayLock锁的wait情况
Expect      :
    1. 创建表，成功
    2. 创建存储过程，成功
    3. 初始查询ProcArrayLock锁的wait情况，正常
    4. 并发执行存储过程，开始执行
    5. kill-9执行事务，kill成功
    6. 再次查询ProcArrayLock锁的wait情况，较步骤3不变
    7. 验证并发事务执行结果，结果正确
    8. 并发执行存储过程，执行正常
    9. 验证并发事务执行结果，结果正确
    10. 再次查询ProcArrayLock锁的wait情况，较步骤6不变
History     :
    modified:2024-3-7 by 5328113; 增加线程超时等待时间，以防执行过慢导致失败
    modified:2024-3-18 by 5328113; 增加线程超时等待时间，以防执行过慢导致失败
"""
import os
import time
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node


class TestCase(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)}:start----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.pri_dbuser = Node(node='PrimaryDbUser')
        self.constant = Constant()
        self.com = Common()
        self.t_name = 't_subxact_case0005'
        self.proc_name = 'proc_subxact_case0005'
        self.thread_num = 20
        self.errflag = 'ERROR:  canceling statement due to user request'

    def test_main(self):
        step_txt = '----step1:创建表; expect:创建成功----'
        self.log.info(step_txt)
        sql = f"drop table if exists {self.t_name};" \
            f"create table {self.t_name}(col1 int);"
        create_result = self.pri_sh.executDbSql(sql)
        self.log.info(create_result)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, create_result,
                      '执行失败:' + step_txt)

        step_txt = '----step2:创建创建存储过程; expect:创建成功----'
        self.log.info(step_txt)
        sql = f'''
            create or replace procedure {self.proc_name}()
            as
            begin
              for i in 1..100000 loop
                  insert into {self.t_name}(col1) values (i);
                if i = 100000 then
                  rollback;
                end if;
              end loop;
            end;
        '''
        create_result = self.pri_sh.executDbSql(sql)
        self.log.info(create_result)
        self.assertIn(self.constant.create_procedure_success_msg,
                      create_result, '执行失败:' + step_txt)

        step_txt = '----step3:初始查询ProcArrayLock锁的wait情况; expect:查询结果正常----'
        self.log.info(step_txt)
        check_sql = f"select * from dbe_perf.wait_events where " \
            f"event='ProcArrayLock';"
        select_result = self.pri_sh.executDbSql(check_sql)
        self.log.info(select_result)
        wait_res = self.com.parse_sql_res_with_col_name(select_result, False)
        wait_1 = int(wait_res[0]['wait'])
        self.assertGreaterEqual(wait_1, 0, '执行失败:' + step_txt)

        step_txt = '----step4:并发执行存储过程; expect:开始执行----'
        self.log.info(step_txt)
        call_thread = []
        call_result = []
        call_sql = f"call {self.proc_name}();"
        for j in range(self.thread_num):
            call_thread.append(
                ComThread(self.pri_sh.executDbSql, args=(call_sql,)))
            call_thread[j].setDaemon(True)
            call_thread[j].start()

        time.sleep(3)
        step_txt = '----step5:kill-9执行事务； expect:kill成功----'
        self.log.info(step_txt)
        self.com.kill_pid_keyword(self.proc_name, 9, self.pri_dbuser)
        for j in range(self.thread_num):
            call_thread[j].join(50)
            result = call_thread[j].get_result()
            call_result.append(result)
        if len(set(call_result)) != 1:
            self.log.info(call_result)
        self.assertNotIn('error', '\n'.join(call_result).lower(),
                         '执行失败:' + step_txt)

        step_txt = '----step6:再次查询ProcArrayLock锁的wait情况; expect:较步骤3不变----'
        self.log.info(step_txt)
        select_result = self.pri_sh.executDbSql(check_sql)
        self.log.info(select_result)
        wait_res = self.com.parse_sql_res_with_col_name(select_result, False)
        wait_2 = int(wait_res[0]['wait'])
        self.assertEqual(wait_2, wait_1, '执行失败:' + step_txt)

        step_txt = '----step7:验证并发事务执行结果; expect:结果正确----'
        self.log.info(step_txt)
        sql = f"select count(*) from {self.t_name};"
        select_result = self.pri_sh.executDbSql(sql)
        self.log.info(select_result)
        count_result = int(select_result.splitlines()[-2].strip())
        self.assertEqual(count_result, 0, '执行失败:' + step_txt)

        step_txt = '----step8:并发执行存储过程; expect:执行正常----'
        self.log.info(step_txt)
        call_thread = []
        call_result = []
        call_sql = f"call {self.proc_name}();"
        for j in range(self.thread_num):
            call_thread.append(
                ComThread(self.pri_sh.executDbSql, args=(call_sql,)))
            call_thread[j].setDaemon(True)
            call_thread[j].start()
        self.log.info(time.asctime())
        for j in range(self.thread_num):
            call_thread[j].join(200)
            result = call_thread[j].get_result()
            call_result.append(result)
        if len(set(call_result)) != 1:
            self.log.info(call_result)
        self.log.info(time.asctime())
        self.assertEqual(len(set(call_result)), 1, '执行失败:' + step_txt)
        self.assertEqual(len(call_result), self.thread_num,
                         '执行失败:' + step_txt)

        step_txt = '----step9:验证并发事务执行结果; expect:结果正确----'
        self.log.info(step_txt)
        sql = f"select count(*) from {self.t_name};"
        select_result = self.pri_sh.executDbSql(sql)
        self.log.info(select_result)
        count_result = int(select_result.splitlines()[-2].strip())
        self.assertEqual(count_result, 0, '执行失败:' + step_txt)

        step_txt = '----step10:再次查询ProcArrayLock锁的wait情况; expect:较步骤6不变----'
        self.log.info(step_txt)
        select_result = self.pri_sh.executDbSql(check_sql)
        self.log.info(select_result)
        wait_res = self.com.parse_sql_res_with_col_name(select_result, False)
        wait_3 = int(wait_res[0]['wait'])
        self.assertEqual(wait_3, wait_2, '执行失败:' + step_txt)

    def tearDown(self):
        step_txt = '----this is teardown----'
        self.log.info(step_txt)
        step1_txt = '----删除表、存储过程; expect:删除成功----'
        self.log.info(step1_txt)
        sql = f"drop table if exists {self.t_name};" \
            f"drop procedure if exists {self.proc_name};"
        drop_result = self.pri_sh.executDbSql(sql)
        self.log.info(drop_result)

        self.log.info(f'----{os.path.basename(__file__)}:end----')
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_result,
                      '执行失败:' + step_txt)
        self.assertIn(self.constant.drop_procedure_success_msg, drop_result,
                      '执行失败:' + step_txt)
